package com.atguigu.day03;

import com.atguigu.utils.IntegerSource;
import com.atguigu.utils.IntegerStatistic;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // ExecutionGraph
        // +--------------------+        +----------------------+
        // | source => map      |  ----> | reduce => print      |
        // +--------------------+        +----------------------+
        env
                .addSource(new IntegerSource())
                .map(r -> new IntegerStatistic(r, r, r, 1, r))
                .returns(Types.POJO(IntegerStatistic.class))
                .keyBy(r -> "int")
                .reduce(new Statistic())
                .print();

        env.execute();
    }

    public static class Statistic implements ReduceFunction<IntegerStatistic> {
        @Override
        public IntegerStatistic reduce(IntegerStatistic accumulator, IntegerStatistic in) throws Exception {
            return new IntegerStatistic(
                    Math.min(accumulator.min, in.min),
                    Math.max(accumulator.max, in.max),
                    accumulator.sum + in.sum,
                    accumulator.count + in.count,
                    (accumulator.sum + in.sum) / (accumulator.count + in.count)
            );
        }
    }
}
